/**
 * File: AbstractQueueProcessor.java
 * Created by: mhaimel
 * Created on: May 5, 2010
 * CVS:  $Id: AbstractQueueProcessor.java 1.0 May 5, 2010 4:07:53 PM mhaimel Exp $
 */
package uk.ac.ebi.curtain.processor;

import java.io.Serializable;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicReference;

import uk.ac.ebi.curtain.utils.concurrent.BlockingQueueShutdown;

/**
 * @author mhaimel
 *
 */
public abstract class AbstractQueueProcessor<I extends Serializable,O extends Serializable> extends AbstractProcessor <I,O>{
	
	private final AtomicReference<BlockingQueueShutdown<I>> inputQueue = new AtomicReference<BlockingQueueShutdown<I>>();
	private final AtomicReference<BlockingQueue<O>> outputQueue = new AtomicReference<BlockingQueue<O>>();


	public void setInputQueue(BlockingQueueShutdown<I> queue) {
		this.inputQueue.set(queue);
	}		
	public BlockingQueueShutdown<I> getInputQueue() {
		return inputQueue.get();
	}
	
	@Override
	public void process() {
		int cnt = 0;
		try {
			while(true){
				if(isForceShutdown()){
					getLog().warn("FORCE shutdown!");
					break;
				}
				I input = takeInput();
				if(null != input){
					O output = doProcess(input);
					submitOutput(output);
				} else if(getInputQueue().isShutdown()){
					getLog().debug("Finished queue!");
					break;// finish processing
				}				
			}			
		} catch (InterruptedException e) {
			getLog().warn("Interrupted!!",e);
			Thread.currentThread().interrupt();
		}
		getLog().debug("Finished processing items: " + cnt);
	}
	
	protected I takeInput() throws InterruptedException {
		return getInputQueue().take();
	}

	protected void submitOutput(O output) {
		if(null != output){
			getOutputQueue().add(output);
		}
	}
	protected abstract O doProcess(I input) throws InterruptedException;
	
	public void setOutputQueue(BlockingQueue<O> queue) {
		this.outputQueue.set(queue);
	}	
	
	public BlockingQueue<O> getOutputQueue() {
		return outputQueue.get();
	}

}
